package zzg.study.reactor;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectableChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Queue;
import java.util.concurrent.LinkedBlockingDeque;

/**
 * @Classname PollerIO
 * @Description TODO
 * @Date 2025/1/27 16:17
 * @Author zhuzhenguo
 */
public class PollerIO extends ServiceThread {

    private final Server server;

    private final Queue<SocketChannel> acceptedQueue;

    public PollerIO(String name, Server server) {
        super(name);
        this.server = server;
        this.acceptedQueue = new LinkedBlockingDeque<>();
    }

    public void addAcceptedChannel(SocketChannel socketChannel) {
        this.acceptedQueue.offer(socketChannel);
        wakeupSelector();
    }

    @Override
    public void run() {
        while (!server.stopped) {
            doSelect();
            doAcceptedConnections();
        }
        closeSelector();
    }

    private void doAcceptedConnections() {
        SocketChannel socketChannel;
        while ((socketChannel = acceptedQueue.poll()) != null && !server.stopped) {
            try {
                socketChannel.register(this.selector, SelectionKey.OP_READ);
            } catch (ClosedChannelException e) {
                throw new RuntimeException(e);
            }
        }
    }

    private void doSelect() {
        try {
            this.selector.select();
            Iterator<SelectionKey> iterator = this.selector.selectedKeys().iterator();
            while (iterator.hasNext()) {
                SelectionKey key = iterator.next();
                iterator.remove();
                if (!key.isValid()) {
                    continue;
                }
                if (key.isReadable()) {
                    handleIO(key);
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private void handleIO(SelectionKey key) {
        if (key.isReadable()) {
            SocketChannel socketChannel = (SocketChannel)key.channel();
            ByteBuffer readBuffer = ByteBuffer.allocate(256);
            try {
                int i = socketChannel.read(readBuffer);
                if (i > 0) {
                    // 将数据封装成任务给线程池做业务
                    BuTask task = new BuTask(readBuffer, socketChannel);
                    server.addBuTask(task);
                }
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }
    }
}
